package cn.doitedu.marketing.engine.service;

import cn.doitedu.marketing.engine.beans.EventBean;
import cn.doitedu.marketing.engine.beans.EventCombinationCondition;
import cn.doitedu.marketing.engine.beans.MarketingRule;
import cn.doitedu.marketing.engine.buffer.BufferManager;
import cn.doitedu.marketing.engine.gauges.CountGauge;
import cn.doitedu.marketing.engine.gauges.QueryGauges;
import cn.doitedu.marketing.engine.querier.ClickHouseQuerier;
import cn.doitedu.marketing.engine.querier.HbaseQuerier;
import cn.doitedu.marketing.engine.querier.StateQuerier;
import cn.doitedu.marketing.engine.utils.CrossTimeQueryUtil;
import cn.doitedu.marketing.engine.utils.EventUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.state.ListState;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/7/26
 **/

@Slf4j
public class TriggerModelConditionQueryService {

    ClickHouseQuerier clickHouseQuerier;
    StateQuerier stateQuerier;
    HbaseQuerier hbaseQuerier;
    BufferManager bufferManager;
    CountGauge bufferHitCountGauge;

    public TriggerModelConditionQueryService(Connection hbaseConn, java.sql.Connection ckConn, ListState<EventBean> beansState, QueryGauges queryGauges, BufferManager bufferManager) throws Exception {
        clickHouseQuerier = new ClickHouseQuerier(ckConn, queryGauges);
        hbaseQuerier = new HbaseQuerier(hbaseConn, queryGauges);
        stateQuerier = new StateQuerier(beansState);
        this.bufferManager = bufferManager;
        this.bufferHitCountGauge = queryGauges.getBufferHitCountGauge();

    }

    private boolean matchEventCombinationCondition(EventCombinationCondition eventCombinationCondition, String deviceId, EventBean event, long timeRangeStart, long timeRangeEnd) throws Exception {
        long segmentPoint = CrossTimeQueryUtil.getSegmentPoint(event.getTimeStamp());

        // state查询
        if (eventCombinationCondition.getTimeRangeStart() >= segmentPoint) {

            long bufferValue = bufferManager.getLong(deviceId, eventCombinationCondition, timeRangeStart, timeRangeEnd);
            if (bufferValue >= eventCombinationCondition.getMinLimit()) {
                log.debug("buffer缓存命中，查询返回");
                bufferHitCountGauge.inc();
                return true;
            }

            log.debug("条件区间完全落在state中，开始state查询......");
            int count = stateQuerier.findEventCombinationConditionCount(deviceId, eventCombinationCondition, eventCombinationCondition.getTimeRangeStart(), eventCombinationCondition.getTimeRangeEnd());
            bufferManager.put(deviceId,eventCombinationCondition,timeRangeStart,timeRangeEnd,count+"");
            if (count < eventCombinationCondition.getMinLimit()) return false;

        }
        // clickhouse 查询
        else if (eventCombinationCondition.getTimeRangeEnd() < segmentPoint) {

            long bufferValue = bufferManager.getLong(deviceId, eventCombinationCondition, timeRangeStart, timeRangeEnd);
            if (bufferValue >= eventCombinationCondition.getMinLimit()) {
                log.debug("buffer缓存命中，查询返回");
                bufferHitCountGauge.inc();
                return true;
            }

            log.debug("条件区间完全落在clickhouse中，开始clickhouse查询......");
            int count = clickHouseQuerier.findEventCombinationConditionCount(eventCombinationCondition, event.getDeviceId(), eventCombinationCondition.getTimeRangeStart(), eventCombinationCondition.getTimeRangeEnd(), false);
            bufferManager.put(deviceId,eventCombinationCondition,timeRangeStart,timeRangeEnd,count+"");
            if (count < eventCombinationCondition.getMinLimit()) return false;
        }
        // 跨区间查询
        else {
            long bufferValue = bufferManager.getLong(deviceId, eventCombinationCondition, timeRangeStart, timeRangeEnd);
            if (bufferValue >= eventCombinationCondition.getMinLimit()) {
                log.debug("buffer缓存命中，查询返回");
                bufferHitCountGauge.inc();
                return true;
            }

            log.debug("条件区间横跨分界点，开始跨区间查询......");
            int count = 0;
            // 先查state
            bufferValue = bufferManager.getLong(deviceId, eventCombinationCondition, segmentPoint, timeRangeEnd);
            if (bufferValue >= eventCombinationCondition.getMinLimit()) {
                log.debug("buffer缓存命中，查询返回");
                bufferHitCountGauge.inc();
                return true;
            }
            count = stateQuerier.findEventCombinationConditionCount(event.getDeviceId(), eventCombinationCondition, segmentPoint, eventCombinationCondition.getTimeRangeEnd());
            if (count >= eventCombinationCondition.getMinLimit()) {
                log.debug("本跨区间条件在state中计算匹配通过，直接跳入下一条件计算");
                bufferManager.put(deviceId,eventCombinationCondition,segmentPoint,timeRangeEnd,count+"");
                return true;
            }

            log.debug("本跨区间条件在state中计算失败，开始跨区间计算");
            String str1 = clickHouseQuerier.getEventCombinationStr(eventCombinationCondition, event.getDeviceId(), eventCombinationCondition.getTimeRangeStart(), segmentPoint);
            String str2 = stateQuerier.getEventCombinationStr(eventCombinationCondition, segmentPoint, eventCombinationCondition.getTimeRangeEnd());
            log.debug("跨区间查询，正则：{} ,拼接事件字符串，ck中返回：{} , state中返回：{}", eventCombinationCondition.getMatchPattern(), str1, str2);
            count = EventUtil.eventStrMatchGroupCount(str1 + str2, eventCombinationCondition.getMatchPattern());
            bufferManager.put(deviceId,eventCombinationCondition,timeRangeStart,timeRangeEnd,count+"");
            if (count < eventCombinationCondition.getMinLimit()) {
                log.debug("跨区间条件，跨区间计算失败，匹配总次数：{}, 条件阈值：{}", count, eventCombinationCondition.getMinLimit());
                return false;
            }
            log.debug("本跨区间条件，跨区间计算匹配成功!");
        }

        return true;
    }

    public boolean matchEventCombinationCondition(EventCombinationCondition eventCombinationCondition, String deviceId, EventBean event) throws Exception {

        boolean b = matchEventCombinationCondition(eventCombinationCondition,
                deviceId,
                event,
                eventCombinationCondition.getTimeRangeStart(),
                eventCombinationCondition.getTimeRangeEnd());
        return b;
    }


    public boolean matchProfileCondition(String deviceId, MarketingRule rule) throws IOException {
        boolean b = hbaseQuerier.queryProfileCondition(deviceId, rule.getUserProfileConditions());
        return b;
    }


}
